package com.hubiwei.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hubiwei.gmall.realtime.app.BaseAppV2;
import com.hubiwei.gmall.realtime.bean.OrderDetail;
import com.hubiwei.gmall.realtime.bean.OrderInfo;
import com.hubiwei.gmall.realtime.bean.OrderWide;
import com.hubiwei.gmall.realtime.common.Constant;
import com.hubiwei.gmall.realtime.util.DimUtil;
import com.hubiwei.gmall.realtime.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.HashMap;

import static com.hubiwei.gmall.realtime.common.Constant.TOPIC_DWD_ORDER_DETAIL;
import static com.hubiwei.gmall.realtime.common.Constant.TOPIC_DWD_ORDER_INFO;


public class DwmOrderWideApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp().init(3003, 1, "DwmOrderWideApp", "DwmOrderWideApp",
                                   TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
    }
    
    @Override
    public void run(StreamExecutionEnvironment env,
                    HashMap<String, DataStreamSource<String>> topicAndStreamMap) {
        // 1. 事实表的join: interval join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = joinFacts(topicAndStreamMap);
        // 2. join维度数据
        joinDims(orderWideStreamWithoutDims);
        
        // 3. 把结果写入到kafka中, 给dws准备数据
    }
    
    private void joinDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
        // join 6张维度表
        // 根据维度表的中的某个id去查找对应的那一条数据
        orderWideStreamWithoutDims
            .map(new RichMapFunction<OrderWide, OrderWide>() {
                
                private Connection phoenixConnection;
                @Override
                public void open(Configuration parameters) throws Exception {
                    phoenixConnection = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
                }
                
                @Override
                public OrderWide map(OrderWide orderWide) throws Exception {
                    // 1. 补充user
                    JSONObject userInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                    "dim_user_info",
                                                                    orderWide.getUser_id());
                    orderWide.setUser_gender(userInfo.getString("GENDER"));
                    orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
                    
                    // 2. 补齐省份
                    JSONObject provinceInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                        "dim_base_province",
                                                                        orderWide.getProvince_id());
                    orderWide.setProvince_name(provinceInfo.getString("NAME"));
                    orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
                    orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
                    orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
                    
                    // 3. sku
                    JSONObject skuInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                        "dim_sku_info",
                                                                        orderWide.getSku_id());
                    orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                    
                    orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                    orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                    orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                    
                    // 4. spu
                    JSONObject spuInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                  "dim_spu_info",
                                                                  orderWide.getSpu_id());
                    orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
                    
                    
                    // 5. tm
                    JSONObject tmInfo = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                  "dim_base_trademark",
                                                                  orderWide.getTm_id());
                    orderWide.setTm_name(tmInfo.getString("TM_NAME"));
                    
                    // 5. tm
                    JSONObject c3Info = DimUtil.getDimFromPhoenix(phoenixConnection,
                                                                  "dim_base_category3",
                                                                  orderWide.getCategory3_id());
                    orderWide.setCategory3_name(c3Info.getString("NAME"));
                    
                    return orderWide;
                }
            })
            .print();
    }
    
    private SingleOutputStreamOperator<OrderWide> joinFacts(HashMap<String, DataStreamSource<String>> topicAndStreamMap) {
        KeyedStream<OrderInfo, Long> orderInfoStream = topicAndStreamMap
            .get(TOPIC_DWD_ORDER_INFO)
            .map(info -> JSON.parseObject(info, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((info, ts) -> info.getCreate_ts())
            )
            .keyBy(OrderInfo::getId);
        KeyedStream<OrderDetail, Long> orderDetailStream = topicAndStreamMap
            .get(TOPIC_DWD_ORDER_DETAIL)
            .map(info -> JSON.parseObject(info, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
            )
            .keyBy(OrderDetail::getOrder_id);
        
        return orderInfoStream
            .intervalJoin(orderDetailStream)
            .between(Time.seconds(-10), Time.seconds(10))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });
        
    }
}
/*,
从kafka读取几个topic:
    1. dwd_order_info
    2. dwd_order_detail
    
    
根据维度的id去join对应维度信息

 */